package com.hitqz.robot.biz.websocket.endpoint;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.nacos.shaded.com.google.common.collect.Maps;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hitqz.robot.api.business.dto.SubscriptTopicDto;
import com.hitqz.robot.biz.websocket.config.CustomConfigurator;
import com.hitqz.robot.common.security.service.RobotUser;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 *  topic : summary 机器人信息
 *  topic: jobCaches 缓存的任务信息
 */
@ServerEndpoint(value = "/websocket/subscribe/{topic}",configurator = CustomConfigurator.class)
@Component
@Slf4j
public class SubscriptionEndpoint {
	// ObjectMapper for JSON parsing
	private static final ObjectMapper objectMapper = new ObjectMapper();
	// 存储话题和session的映射关系
	public static final Map<String, Set<Session>> topicSubscribers = new ConcurrentHashMap<>();
	// 用于存储所有打开的WebSocket sessions
	private static final Set<Session> sessions = new HashSet<>();

	@OnOpen
	public void onOpen(Session session,@PathParam(value = "topic") String topic) {
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		sessions.add(session);
		log.debug("[ws][subscribe]{},session-id:{} connected.current size :{}", user.getName(),session.getId(),sessions.size());
		//订阅话题
		subscribe(session, topic);
	}

	// 订阅话题   客户端需要发送一个JSON消息，格式为：{"action":"subscribe","topic":"topicName"}
	@OnMessage
	public void onMessage(Session session,String message) throws IOException {
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		//log.info("[ws][subscribe]{}, on message:{},session-id:{}", user.getName(),message,session.getId());
		if (message.contains("ping")) {
			Map<String, Object> ret = Maps.newHashMap();
			ret.put("type", "pong");
			session.getAsyncRemote().sendText(objectMapper.writeValueAsString(ret));
		}
	}

	@OnClose
	public void onClose(Session session,@PathParam(value = "topic") String topic) {
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		//取消订阅
		unsubscribe(session, topic);
		// 移除session
		sessions.remove(session);
		log.debug("[ws][subscribe]{},session-id:{} disconnected.current size :{}", user.getName(),session.getId(),sessions.size());
	}

	@OnError
	public void onError(Session session,Throwable error) {
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		log.error("[ws][subscribe]{},session-id:{},error:{}", user.getName(),session.getId(),error.getMessage());
	}

	// 话题广播
	public static void broadcast(String topic, String message) {
		if (topicSubscribers.containsKey(topic)) {
			Set<Session> subscribers = topicSubscribers.get(topic);
			if (subscribers.isEmpty()) {
				topicSubscribers.remove(topic);
				return;
			}
			// 清理已关闭的订阅者
			subscribers.removeIf(subscriber -> !subscriber.isOpen());
			for (Session subscriber : topicSubscribers.get(topic)) {
				if (subscriber.isOpen()) {
					subscriber.getAsyncRemote().sendText(message);
				}
			}
		}
	}

	// 订阅话题
	public static void subscribe(Session session, String topic) {
		topicSubscribers.computeIfAbsent(topic, k -> new HashSet<>()).add(session);
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		log.info("[ws][subscribe]{},session-id:{} subscribe topic:{}",user.getName(),session.getId(),topic);
		SpringUtil.getApplicationContext().publishEvent(new SubscriptTopicDto(topic));
	}

	// 取消订阅
	public static void unsubscribe(Session session, String topic) {
		// 移除订阅session
		if (topicSubscribers.containsKey(topic)) {
			topicSubscribers.get(topic).remove(session);
			// 进一步移除话题
			if (topicSubscribers.get(topic).isEmpty()) {
				topicSubscribers.remove(topic);
			}
			RobotUser user =(RobotUser) session.getUserProperties().get("user");
			log.info("[ws][subscribe]{},session-id:{} unsubscribe topic:{}",user.getName(),session.getId(),topic);
		}
	}

	private void sendError(Session session, String errorMessage) {
		RobotUser user =(RobotUser) session.getUserProperties().get("user");
		try {
			session.getBasicRemote().sendText("{\"error\":\"" + errorMessage + "\"}");
		} catch (IOException e) {
			// 处理发送错误消息时的异常
			log.error("[ws][subscribe]send to {} error message failed:{}",user.getName(),e.getMessage());
		}finally {
			try {
				// 在发送错误信息后关闭会话
				session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT, errorMessage));
			} catch (IOException e) {
				// 处理关闭session时的异常
				log.error("[ws][subscribe]close {}, session-id:{} failed:{}",user.getName(),session.getId(),e.getMessage());
			}
		}
	}

}
